package Outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * Created by yqz on 2019/4/12.
 */
public class FilterLogRecordWriter extends RecordWriter<Text, NullWritable> {
    FSDataOutputStream itstarOut = null;
    FSDataOutputStream otherOut = null;

    /**
     * 数据的输出的环节
     *
     * @param job 用于获取配置信息，从而得到fs
     */
    public FilterLogRecordWriter(TaskAttemptContext job) {

        try {
            FileSystem fs = FileSystem.get(job.getConfiguration());

            Path itstarPath = new Path("F:\\date\\A\\itstarlog\\itstar.log");
            Path otherPath = new Path("F:\\date\\A\\itstarlog\\other.log");

            itstarOut = fs.create(itstarPath);
            otherOut = fs.create(otherPath);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {
        if (text.toString().contains("itstar")) {
            itstarOut.write(text.toString().getBytes());
        } else {
            otherOut.write(text.toString().getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        if (itstarOut != null) {
            itstarOut.close();
        }
        if (otherOut != null) {
            otherOut.close();
        }
    }
}
